﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections.Concurrent;
using PServiceBus.Core.Runtime;
using PServiceBus.Core.Logger;
using System.Threading.Tasks;
using System.Collections;
using PServiceBus.Core.Runtime.Extensions;
using Cassandraemon;
using Apache.Cassandra;
using Thrift.Protocol;

namespace PServiceBus.Cassandra {
    /// <summary>
    /// Base class for cassandra data store
    /// e.x connectionstring: endpoint://guest:guest@localhost:9160/hosts=localhost2&keyspace=psb&cf=psb.messagestore&rf=1&sc=SimpleStrategy&clr=quorum&clw=one
    /// </summary>
    /// <typeparam name="TObject"></typeparam>
    public abstract class CassandraCollection<TObject> : IDisposable, IEnumerable<TObject> where TObject : class {

        private object _lockQueryObject = new object();
        private bool _isDisposed;
        private const string CASSANDRA_PERSIST_KEY = "QueryPersist";
        private const string DEFAULT_KEYSPACE = "KS_PSB";
        protected const string READ_ONLY_KEY = "ReadOnly";
        

        private ConcurrentDictionary<string, CassandraContext> _clients = new ConcurrentDictionary<string, CassandraContext>();
        private ConcurrentDictionary<string, bool> _clientStates = new ConcurrentDictionary<string, bool>();

        public string Endpoint { get; set; }

        public string EndpointAlias { get; set; }

        public bool AppendTypeNameToEndpoint { get; set; }

        public Func<TObject, Guid> IDFunc { get; set; }

        protected byte[] GetID(TObject obj) {
            return IDFunc(obj).ToCassandraByte();
        }

        private EndpointUri GetEndpointUri() {
            var endpoint = Endpoint;
            var tokens = (AppendTypeNameToEndpoint ? typeof(TObject).FullName : string.Empty).Split('.');
            var path = string.Empty;
            for (var i = 0; i < tokens.Length; i++) {
                var token = tokens[i];
                if (String.IsNullOrWhiteSpace(token)) continue;
                if (i == tokens.Length - 1)
                    path += token;
                else {
                    path += token.Length > 2 ? token.Substring(0, 2) : token;
                }
            }
            path += (EndpointAlias ?? string.Empty);
            endpoint += String.Concat((endpoint.EndsWith("&") || String.IsNullOrWhiteSpace(endpoint) ? string.Empty : "&"), "cf=", path);
            return EndpointUri.Parse(endpoint);
        }

        private CassandraContext GetCassandraConnection(string name) {
            var info = CassandraInfo;
            var keyspace = CassandraPathInfo.KeySpace;
            var client = new CassandraContext(info.Host, info.Port, "system", 
                name == CASSANDRA_PERSIST_KEY ? CassandraPathInfo.WriteConsistencyLevel : CassandraPathInfo.ReadConsistencyLevel);

            if (!String.IsNullOrWhiteSpace(info.Password) && !String.IsNullOrWhiteSpace(info.Username))
                client.Login(new Dictionary<string, string> { {"username", info.Username}, {"password", info.Password} });

            //Change to current keyspace
            var keySpaceObjs = client.DescribeKeySpaces();
            var keyspaceObj = keySpaceObjs.FirstOrDefault(x => x.Name.ToLower() == keyspace.ToLower());

            if (keyspaceObj == null) {
                client.SystemAddKeyspace(new Apache.Cassandra.KsDef()
                {
                    Name = keyspace,
                    Durable_writes = true,
                    Strategy_class = CassandraPathInfo.StrategyClass,
                    Strategy_options = new Dictionary<string, string> { { "replication_factor", CassandraPathInfo.ReplicationFactor.ToString() } },
                    Cf_defs = new List<CfDef>()
                });
            } else {
                client.Close();
                client = new CassandraContext(info.Host, info.Port, keyspace,
                name == CASSANDRA_PERSIST_KEY ? CassandraPathInfo.WriteConsistencyLevel : CassandraPathInfo.ReadConsistencyLevel);
                if (!String.IsNullOrWhiteSpace(info.Password) && !String.IsNullOrWhiteSpace(info.Username))
                    client.Login(new Dictionary<string, string> { { "username", info.Username }, { "password", info.Password } });
            }

            
            //Prepare column family for current collection object
            PrepareColumnFamily(CassandraPathInfo.ColumnFamily, client);

            return client;
        }

        protected CassandraContext GetConnection(string name) {
            var client = default(CassandraContext);
            var state = false;
            var info = CassandraInfo;

            if (!_clientStates.TryGetValue(name, out state)) {
                state = _clientStates[name] = false;
            }

            if (!_clients.TryGetValue(name, out client)) {
                client = _clients[name] = GetCassandraConnection(name);
            }
            //state will be true if client is disposed
            if (client == null || state) {
                MethodHelper.Try(() => client.Dispose());
                client = _clients[name] = GetCassandraConnection(name);
            }

            client.Disposed += (s, e) => {
                _clientStates[name] = true;
            };

            return client;
        }


        protected void Query(Action<CassandraContext> action) {
            try {
                lock (_lockQueryObject) {
                    var client = GetConnection(CASSANDRA_PERSIST_KEY);
                    action(client);
                    client.SubmitChanges();
                }
            } catch (Exception ex) {
                ESBLogger.Log(ex);
            }
        }

        protected TReturn Query<TReturn>(Func<CassandraContext, TReturn> func) {
            var result = default(TReturn);
            try {
                lock (_lockQueryObject) {
                    result = func(GetConnection(CASSANDRA_PERSIST_KEY));
                }
            } catch (Exception ex) {
                ESBLogger.Log(ex);
            }
            return result;
        }

        protected bool PrepareColumnFamily(string name, CassandraContext client, Action<Apache.Cassandra.CfDef> action = null) {
            if (String.IsNullOrWhiteSpace(name))
                return false;

            var isNew = false;
            var keyspaceObj = client.DescribeKeySpace(CassandraPathInfo.KeySpace);
            if (!keyspaceObj.Cf_defs.Any(x => x.Name == name)) {
                var cf = new Apache.Cassandra.CfDef()
                {
                    Name = name,
                    Keyspace = CassandraPathInfo.KeySpace
                };
                if (action != null)
                    action(cf);
                client.SystemAddColumnFamily(cf);
                isNew = true;
            }
            return isNew || (keyspaceObj == null);
        }


        private ulong CounterIncrDecr(string key, bool incr = true) {
            return Query(client =>
            {
                var counterFamily = CassandraConstants.COUNTER_COLUMN_FAMILY;
                var isNew = PrepareColumnFamily(counterFamily, client, cf =>
                {
                    cf.Default_validation_class = "CounterColumnType";
                });
                var value = isNew ? (incr ? 1 : 0) : (incr ? 1 : -1);

                client.CounterColumn.InsertOnSubmit(new CassandraEntity<CounterColumn>().SetColumnFamily(counterFamily).SetKey(counterFamily).SetData((long)value, key));
                client.SubmitChanges();

                return (ulong)client.CounterColumnList.Where(x => x.ColumnFamily == counterFamily && x.Key == counterFamily
                    && x.Column == key).First().Data.First().Value;
            });
        }

        public ulong Incr(string key) {
            return CounterIncrDecr(key);
        }

        public ulong Decr(string key) {
            return CounterIncrDecr(key, incr: false);
        }

        public TValue Get<TValue>(string key) {
            return Query(client =>
            {
                var kvFamily = CassandraConstants.KEYVALUES_COLUMN_FAMILY;
                PrepareColumnFamily(kvFamily, client);
                var data = client.ColumnList.Where(x => x.ColumnFamily == kvFamily && x.Key == kvFamily && x.Column == key).ToList();
                var first = data.FirstOrDefault();

                byte[] buffer = default(byte[]);
                if (first != null && first.Data.Any())
                    buffer = first.Data.First().Value;
                
                return (TValue)Convert.ChangeType(buffer == null ? "0" : buffer.UTF8ByteToString(),
                    typeof(TValue));
            });
        }

        public void Set<TValue>(string key, TValue value) {
            Query(client => {
                var kvFamily = CassandraConstants.KEYVALUES_COLUMN_FAMILY;
                PrepareColumnFamily(kvFamily, client);
                client.Column.InsertOnSubmit(new CassandraEntity<Column>().SetColumnFamily(
                    kvFamily).SetKey(kvFamily).SetData(value.ToString().ToUTF8Bytes(), key));
            });
        }

        protected class CassandraConnectionInfo {
            public string KeySpace { get; set; }
            public string[] Hosts { get; set; }
            public string Prefix { get; set; }
            public string ColumnFamily { get; set; }
            /// <summary>
            /// ColumnFamily will default to Queue name when used in queue mode where queues are stores as single row with multiple columns
            /// </summary>
            public string Queue { get { return ColumnFamily; } }
            public int ReplicationFactor { get; set; }
            public string StrategyClass { get; set; }
            public ConsistencyLevel ReadConsistencyLevel { get; set; }
            public ConsistencyLevel WriteConsistencyLevel { get; set; }
        }

        private CassandraConnectionInfo GetCassandraConnectionInfo(string host, string path) {
            var info = new CassandraConnectionInfo() { KeySpace = DEFAULT_KEYSPACE, 
                ReadConsistencyLevel = ConsistencyLevel.QUORUM, WriteConsistencyLevel = ConsistencyLevel.ONE, ReplicationFactor = 1, 
                StrategyClass = "org.apache.cassandra.locator.SimpleStrategy" };
            var tokens = path.Split('&');
            foreach (var token in tokens) {
                var pair = token.Split('=');
                var key = pair[0].ToLower();
                var value = pair.Length > 1 ? pair[1] : string.Empty;
                
                if (key == "keyspace")
                    info.KeySpace = value;
                else if (key == "cf")
                    info.ColumnFamily = value;
                else if (key == "rf")
                    info.ReplicationFactor = Convert.ToInt32(value);
                else if (key == "sc")
                    info.StrategyClass = "org.apache.cassandra.locator." + value;
                else if (key == "clr")
                    info.ReadConsistencyLevel = (ConsistencyLevel)Enum.Parse(typeof(ConsistencyLevel), value.ToUpper());
                else if (key == "clw")
                    info.WriteConsistencyLevel = (ConsistencyLevel)Enum.Parse(typeof(ConsistencyLevel), value.ToUpper());
                else if (key == "hosts")
                    info.Hosts = value.Replace(";", ",").Split(',').Union(new List<string> { host }).ToArray();
                else if (key == "pfx")
                    info.Prefix = value;
            }
            info.ColumnFamily += info.Prefix ?? String.Empty;
            return info;
        }

        private CassandraConnectionInfo _cassandraConnectionInfo;
        protected CassandraConnectionInfo CassandraPathInfo { 
            get {
                return _cassandraConnectionInfo ?? (_cassandraConnectionInfo = GetCassandraConnectionInfo(CassandraInfo.Host, CassandraInfo.Path));
            } 
        }

        private EndpointUri _endpointUri = null;
        protected EndpointUri CassandraInfo {
            get {
                return _endpointUri ?? (_endpointUri = GetEndpointUri());
            }
        }

        public virtual void Clear() {
            Query(client =>
            {
                client.Truncate(CassandraPathInfo.ColumnFamily);
            });
        }

        internal void ClearColumnFamilies(params string[] cfs) {
            Query(client =>
            {
                foreach (var cf in cfs) 
                    client.Truncate(cf);
            });
        }

        public virtual void Delete() {
            Query(client =>
            {
                client.SystemDropColumnFamily(CassandraPathInfo.ColumnFamily);
            });
        }

        public void Create() {
            Query(client =>
            {
                PrepareColumnFamily(CassandraPathInfo.ColumnFamily, client);
            });
        }

        public abstract IEnumerator<TObject> GetObjects();

        public abstract bool Add(IEnumerable<TObject> list);
        public abstract bool Add(TObject obj);

        public abstract bool Delete(IEnumerable<Func<TObject, bool>> conditions);
        public abstract bool Delete(IEnumerable<TObject> list);
        public abstract bool Delete(Func<TObject, bool> condition);
        public abstract bool Delete(TObject obj);
        public abstract bool Exists(Func<TObject, bool> condition);
        public abstract bool Exists(TObject obj);
        public abstract TObject Get(Func<TObject, bool> condition);
        public abstract TObject Get(Guid id);
        public abstract bool Refresh(TObject obj);

        #region IDisposable Members

        public void Dispose() {
            if (_isDisposed) return;
            foreach (var kv in _clients) {
                var client = kv.Value;
                if (!_clientStates[kv.Key])
                    MethodHelper.Try(client.Dispose);
            }
            _clients.Clear();
            _clientStates.Clear();
            _isDisposed = true;
        }

        #endregion

        #region IEnumerable<TObject> Members

        public IEnumerator<TObject> GetEnumerator() {
            return GetObjects();
        }

        #endregion

        #region IEnumerable Members

        IEnumerator IEnumerable.GetEnumerator() {
            return GetObjects();
        }

        #endregion
    }
}
